package net.bw.realtime.game.job;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-13
 */

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class MongoDBToKafkaJob {

    public static void main(String[] args) {

        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv() ;

        // 2. 输入表-input：映射到mongo消息队列
        createInputTable(tabEnv);

        // 3. 输出表-output：映射到kafka表
        createOutputTable(tabEnv) ;

        // 4. 保存数据-insert
        saveTokafka(tabEnv) ;

    }

    // todo 4.保存数据-insert
    private static void saveTokafka(TableEnvironment tabEnv) {

        tabEnv.executeSql("INSERT INTO game_sales_fact (\n" +
                "  sale_id\n" +
                ")\n" +
                "SELECT\n" +
                "  sale_id\n" +
                "FROM game_sales");

    }

    // todo 3.创建输出表
    private static void createOutputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql("CREATE TABLE game_sales_fact\n" +
                "(\n" +
                "  sale_id     INT,\n" +
                "  PRIMARY KEY (sale_id) NOT ENFORCED  \n" +
                ")\n" +
                "WITH (\n" +
                "      'connector' = 'upsert-kafka',\n" +
                "      'topic' = 'game_sales_fact',\n" +
                "      'properties.bootstrap.servers' = 'node101:9092',\n" +
                "      'key.format' = 'json',\n" +
                "      'value.format' = 'json'\n" +
                ")");

		//tabEnv.executeSql("select * From game_sales limit 20").print();
    }

    // todo 2.创建输入表
    private static void createInputTable(TableEnvironment tabEnv) {

        //创建MongoDB游戏销售表
        tabEnv.executeSql("CREATE  TABLE game_sales\n" +
                "(\n" +
                "  `_id`       STRING ,    \n" +
                "  sale_id     INT,       \n" +
                "  game_id     INT,      \n" +
                "  platform_id INT,      \n" +
                "  sale_date   STRING,   \n" +
                "  units_sold  INT,      \n" +
                "  sale_amt    Double    \n" +
                ")\n" +
                "WITH (\n" +
                " 'connector' = 'mongodb',\n" +
                "   'uri' = 'mongodb://localhost:27017',\n" +
                "   'database' = 'db',\n" +
                "   'collection' = 'game_sales'\n" +
                ")");

		//tabEnv.executeSql("select * From game_sales limit 20").print();
    }

    // todo 1.表的执行环境设置
    private static TableEnvironment getTableEnv(){

        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tabEnv = TableEnvironment.create(settings) ;

        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        // 3返回对象
        return tabEnv;
    }

}
